package com.beiding.dbrequest.server;

import com.beiding.dbrequest.common.HessianDecoder;
import com.beiding.dbrequest.common.HessianEncoder;
import com.beiding.dbrequest.common.Terminal;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.ServerSocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import io.reactivex.Observable;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.java.Log;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * 服务端封装
 */
@Log
public class Server extends Terminal {


    /**
     * 对客户端暴露的用来连接的服务端通道
     */
    @Getter
    private ServerSocketChannel serverChannel;

    /**
     * 端口
     */
    @Setter
    @Getter
    private Integer port;

    @Getter
    private ChannelSelector channelSelector = new RandomChannelSelector();

    @Getter
    private NioEventLoopGroup bossGroup;

    @Getter
    private NioEventLoopGroup workerGroup;

    @Setter
    @Getter
    private int workerThreadNumber = 0;

    @Setter
    @Getter
    private int bossThreadNumber = 0;

    @Setter
    @Getter
    private Map<String, Object> channelOptionConfig;


    /**
     * 开始服务
     */
    public Observable<?> start() {

        return Observable.create(subscriber -> {

            try {
                synchronized (Server.this) {
                    if (serverChannel != null) {
                        subscriber.onError(new IllegalStateException("服务不可重新启动"));
                        return;
                    }

                    //启动netty服务
                    bossGroup = new NioEventLoopGroup(bossThreadNumber);
                    workerGroup = new NioEventLoopGroup(workerThreadNumber);
                    ServerBootstrap serverBootstrap = new ServerBootstrap();
                    ServerBootstrap channel = serverBootstrap.group(bossGroup, workerGroup)
                            .channel(NioServerSocketChannel.class);
                    if (channelOptionConfig != null) {
                        channelOptionConfig.forEach((k, v) -> {
                            channel.option(ChannelOption.valueOf(k), v);
                        });
                    }

                    //.childHandler(new LoggingHandler(LogLevel.DEBUG))
                    channel
                            .childHandler(new ChannelInitializer<NioSocketChannel>() {

                                @Override
                                protected void initChannel(NioSocketChannel ch) {
                                    ch.pipeline()
                                            .addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS))
                                            .addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, lengthFieldLength, 0, lengthFieldLength))
                                            .addLast(new LengthFieldPrepender(lengthFieldLength))
                                            .addLast(new HessianEncoder())
                                            .addLast(new HessianDecoder())
                                            .addLast(new H())
                                    ;

                                }


                            });

                    //绑定端口
                    ChannelFuture future = serverBootstrap.bind(port);
                    future.addListener(future1 -> {
                        if (!future1.isSuccess()) {
                            log.warning("服务启动失败");
                            subscriber.onError(future1.cause());
                        } else {
                            log.info("服务已启动");
                            subscriber.onComplete();
                        }
                    });
                }

            } catch (Throwable throwable) {
                subscriber.onError(throwable);
            }


        });


    }


    /**
     * 关闭服务
     */
    public Observable<?> close() {

        return Observable.create(subscriber -> {

            AtomicInteger atomicInteger = new AtomicInteger();


            if (serverChannel != null) {
                atomicInteger.addAndGet(1);
            }
            if (workerGroup != null) {
                atomicInteger.addAndGet(1);
            }
            if (bossGroup != null) {
                atomicInteger.addAndGet(1);
            }
            if (serverChannel != null) {
                serverChannel.closeFuture().addListener(future -> {
                    if (future.isSuccess()) {
                        serverChannel = null;
                        if (atomicInteger.addAndGet(-1) == 0) {
                            subscriber.onComplete();
                        }
                    } else {
                        subscriber.onError(future.cause());
                    }
                });

            }
            // 优雅关闭server线程及相关资源
            if (workerGroup != null) {
                workerGroup.shutdownGracefully().addListener(future -> {
                    if (future.isSuccess()) {
                        workerGroup = null;
                        if (atomicInteger.addAndGet(-1) == 0) {
                            subscriber.onComplete();
                        }
                    } else {
                        subscriber.onError(future.cause());
                    }
                });
            }
            if (bossGroup != null) {
                bossGroup.shutdownGracefully().addListener(future -> {
                    if (future.isSuccess()) {
                        bossGroup = null;
                        if (atomicInteger.addAndGet(-1) == 0) {
                            subscriber.onComplete();
                        }
                    } else {
                        subscriber.onError(future.cause());
                    }
                });
            }

            if (atomicInteger.addAndGet(0) == 0) {
                subscriber.onComplete();
            }

        });
    }

}
